/**
 * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
 * agreements.  See the NOTICE file distributed with this work for additional information regarding
 * copyright ownership.  The ASF licenses this file to you under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with the License.  You may obtain
 * a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed under the License
 * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
 * or implied. See the License for the specific language governing permissions and limitations under
 * the License.
 */
package org.apache.zookeeper.server.quorum;

import org.apache.jute.BinaryInputArchive;
import org.apache.jute.BinaryOutputArchive;
import org.apache.jute.InputArchive;
import org.apache.jute.OutputArchive;
import org.apache.jute.Record;
import org.apache.zookeeper.server.Request;
import org.apache.zookeeper.server.ServerCnxn;
import org.apache.zookeeper.server.ZooTrace;
import org.apache.zookeeper.server.quorum.QuorumPeer.QuorumServer;
import org.apache.zookeeper.server.util.SerializeUtils;
import org.apache.zookeeper.server.util.ZxidUtils;
import org.apache.zookeeper.txn.TxnHeader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.File;
import java.io.IOException;
import java.net.ConnectException;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.nio.ByteBuffer;
import java.nio.IntBuffer;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;

/**
 * This class is the superclass of two of the three main actors in a ZK
 * ensemble: Followers and Observers. Both Followers and Observers share 
 * a good deal of code which is moved into Peer to avoid duplication. 
 */
public class Learner {
    protected static final Logger LOG = LoggerFactory.getLogger(Learner.class);
    /**
     * Socket编程中，TCP_NODELAY选项是用来控制是否开启Nagle算法，该算法是为了提高较慢的广域网传输效率，减小小分组的报文个数，完整描述：
     * 该算法要求一个TCP连接上最多只能有一个未被确认的小分组，在该小分组的确认到来之前，不能发送其他小分组。
     */
    static final private boolean nodelay = System.getProperty("follower.nodelay", "true").equals("true");

    static {
        LOG.info("TCP NoDelay set to: " + nodelay);
    }

    final ConcurrentHashMap<Long, ServerCnxn> pendingRevalidations = new ConcurrentHashMap<Long, ServerCnxn>();
    /**
     * 输出流
     * 通过此输出流向Leader发送数据
     */
    protected BufferedOutputStream bufferedOutput;
    protected Socket sock;
    /**
     * leader input stream
     */
    protected InputArchive leaderIs;
    /**
     * leader output stream
     */
    protected OutputArchive leaderOs;
    /** the protocol version of the leader */
    protected int leaderProtocolVersion = 0x01;
    QuorumPeer self;
    LearnerZooKeeperServer zk;

    public Socket getSocket() {
        return sock;
    }

    public int getPendingRevalidationsCount() {
        return pendingRevalidations.size();
    }

    /**
     * validate a session for a client
     *
     * @param clientId
     *                the client to be revalidated
     * @param timeout
     *                the timeout for which the session is valid
     * @return
     * @throws IOException
     */
    void validateSession(ServerCnxn cnxn, long clientId, int timeout) throws IOException {
        LOG.info("Revalidating client: 0x" + Long.toHexString(clientId));
        ByteArrayOutputStream baos = new ByteArrayOutputStream();
        DataOutputStream dos = new DataOutputStream(baos);
        dos.writeLong(clientId);
        dos.writeInt(timeout);
        dos.close();
        QuorumPacket qp = new QuorumPacket(Leader.REVALIDATE, -1, baos.toByteArray(), null);
        pendingRevalidations.put(clientId, cnxn);
        if (LOG.isTraceEnabled()) {
            ZooTrace.logTraceMessage(LOG,
                                     ZooTrace.SESSION_TRACE_MASK,
                                     "To validate session 0x" + Long.toHexString(clientId));
        }
        writePacket(qp, true);
    }

    /**
     * 向Leader发送一个数据包
     *
     * @param pp 需要发送给Leader的proposal数据包
     * @param flush 是否立即flush
     */
    void writePacket(QuorumPacket pp, boolean flush) throws IOException {
        synchronized (leaderOs) {
            if (pp != null) {
                leaderOs.writeRecord(pp, "packet");
            }
            if (flush) {
                bufferedOutput.flush();
            }
        }
    }

    /**
     * 从Leader读取一个数据包
     *
     * @param pp the packet to be instantiated
     */
    void readPacket(QuorumPacket pp) throws IOException {
        synchronized (leaderIs) {
            leaderIs.readRecord(pp, "packet");
        }

        // 下面都是日志
        long traceMask = ZooTrace.SERVER_PACKET_TRACE_MASK;
        if (pp.getType() == Leader.PING) {
            traceMask = ZooTrace.SERVER_PING_TRACE_MASK;
        }
        if (LOG.isTraceEnabled()) {
            ZooTrace.logQuorumPacket(LOG, traceMask, 'i', pp);
        }
    }

    /**
     * send a request packet to the leader
     *
     * @param request
     *                the request from the client
     * @throws IOException
     */
    void request(Request request) throws IOException {
        ByteArrayOutputStream baos = new ByteArrayOutputStream();
        DataOutputStream oa = new DataOutputStream(baos);
        oa.writeLong(request.sessionId);
        oa.writeInt(request.cxid);
        oa.writeInt(request.type);
        if (request.request != null) {
            request.request.rewind();
            int len = request.request.remaining();
            byte b[] = new byte[len];
            request.request.get(b);
            request.request.rewind();
            oa.write(b);
        }
        oa.close();
        QuorumPacket qp = new QuorumPacket(Leader.REQUEST, -1, baos.toByteArray(), request.authInfo);
        writePacket(qp, true);
    }

    /**
     * 返回我们认为是Leader节点的地址address
     * （主要就是根据当前投票的sid，来找到QuorumServer）
     *
     * Returns the address of the node we think is the leader.
     */
    protected QuorumServer findLeader() {
        QuorumServer leaderServer = null;
        // 找到Leader的sid   Find the leader by id
        long leaderSid = self.getCurrentVote().getId();
        for (QuorumServer s : self.getView().values()) {
            if (s.id == leaderSid) {
                // 确保我们在尝试连接之前，已经有Leader的正确IP地址。Ensure we have the leader's correct IP address before attempting to connect.
                s.recreateSocketAddresses();
                leaderServer = s;
                break;
            }
        }
        if (leaderServer == null) {
            LOG.warn("Couldn't find the leader with id = " + leaderSid);
        }
        return leaderServer;
    }

    /**
     * Establish a connection with the Leader found by findLeader. Retries
     * 5 times before giving up.
     * @param addr - the address of the Leader to connect to.
     * @throws IOException <li>if the socket connection fails on the 5th attempt</li>
     * <li>if there is an authentication failure while connecting to leader</li>
     * @throws ConnectException
     * @throws InterruptedException
     */
    protected void connectToLeader(InetSocketAddress addr, String hostname) throws IOException, InterruptedException {
        sock = new Socket();
        // 设置超时时间(阻塞)
        sock.setSoTimeout(self.tickTime * self.initLimit);

        // 5次重试，每次重试间隔1秒
        for (int tries = 0; tries < 5; tries++) {
            try {
                // 阻塞方式连接
                sock.connect(addr, self.tickTime * self.syncLimit);
                sock.setTcpNoDelay(nodelay);
                break;
            } catch (IOException e) {
                if (tries == 4) {
                    LOG.error("Unexpected exception", e);
                    throw e;
                } else {
                    LOG.warn("Unexpected exception, tries=" + tries + ", connecting to " + addr, e);
                    sock = new Socket();
                    sock.setSoTimeout(self.tickTime * self.initLimit);
                }
            }
            Thread.sleep(1000);
        }

        self.authLearner.authenticate(sock, hostname);

        // 设置Leader相关的input/output stream
        leaderIs = BinaryInputArchive.getArchive(new BufferedInputStream(sock.getInputStream()));
        bufferedOutput = new BufferedOutputStream(sock.getOutputStream());
        leaderOs = BinaryOutputArchive.getArchive(bufferedOutput);
    }

    /**
     * 一旦连接到Leader，执行握手协议为了建立一个 following / observing 连接。
     * 步骤：
     *  1、给Leader发送FOLLOWERINFO或OBSERVERINFO类型的消息。(给Leader服务器发送Follower info，包含last zxid和sid)
     *  2、读取Leader发送的数据
     *  3、如果Leader发来的数据包类型为 LEADERINFO ，则读取Leader的zxid，并根据zxid计算出Leader newEpoch
     *
     * 主要处理的消息：
     * 1、FOLLOWERINFO 或 OBSERVERINFO
     * 2、LEADERINFO
     * @param pktType {@link Leader#FOLLOWERINFO}  或者  {@link Leader#OBSERVERINFO}
     * @return the zxid the Leader sends for synchronization purposes.（n.意图; 目的; 用途; 目标; 情势的需要; 重要意义; 有价值的意义;）
     * @throws IOException
     */
    protected long registerWithLeader(int pktType) throws IOException {
        // 发送 FollowerInfo 或者 ObserverInfo, 包含last zxid 和 sid
        long lastLoggedZxid = self.getLastLoggedZxid();
        QuorumPacket qp = new QuorumPacket();
        qp.setType(pktType);
        // zxid
        qp.setZxid(ZxidUtils.makeZxid(self.getAcceptedEpoch(), 0));
        // sid
        LearnerInfo li = new LearnerInfo(self.getId(), 0x10000);
        ByteArrayOutputStream bsid = new ByteArrayOutputStream();
        BinaryOutputArchive boa = BinaryOutputArchive.getArchive(bsid);
        boa.writeRecord(li, "LearnerInfo");
        qp.setData(bsid.toByteArray());

        // 发送数据包
        writePacket(qp, true);

        // 读取数据包
        readPacket(qp);

        // Leader的数据包中：这个 newEpoch 是当前server共同协商出来的 Epoch   （Leader#getEpochToPropose方法）
        final long newEpoch = ZxidUtils.getEpochFromZxid(qp.getZxid());

        // 如果接收Leader发来的数据包的类型为 LEADERINFO
        if (qp.getType() == Leader.LEADERINFO) {
            // we are connected to a 1.0 server so accept the new epoch and read the next packet
            leaderProtocolVersion = ByteBuffer.wrap(qp.getData()).getInt();
            byte epochBytes[] = new byte[4];
            final ByteBuffer wrappedEpochBytes = ByteBuffer.wrap(epochBytes);
            // 通过Leader zxid计算出的Epoch大于自己的Epoch。
            if (newEpoch > self.getAcceptedEpoch()) {
                // 自己当前currentEpoch  保存到 epochBytes 中。(因为后面要发送 ACKEPOCH 消息)
                wrappedEpochBytes.putInt((int) self.getCurrentEpoch());
                self.setAcceptedEpoch(newEpoch);
            } else if (newEpoch == self.getAcceptedEpoch()) {
                // 既然我们已经给新的Leader发送了ack epoch并且把自己的epoch设置为Leader的Epoch，
                // 那么我们就不能在此ack了。但是我们仍然需要给Leader发送我们的 lastZxid，为了我们能够与Leader进行数据同步，如果Leader确实担任新epoch周期的领导。
                // -1表示Follower发送的这个ACKEPOCH响应，不应该被作为新epoch的ack统计计数中。
                wrappedEpochBytes.putInt(-1);
                // leader.newEpoch理论上来说不可能和self.epoch相等。出现这种情况，可以认为是非正常情况。
                // 疑问:如果newLeader.newEpoch 和learner.epoch相等，为何这里要发送-1呢？
                //  答：传epoch为-1,在 Leader#waitForEpochAck 方法中， 如果ss.getCurrentEpoch()等于-1，就不会把当前Learner sid添加到electingFollowers中了。
            } else {
                throw new IOException("Leaders epoch, " + newEpoch + " is less than accepted epoch, " + self.getAcceptedEpoch());
            }

            // Follower 发送消息：ACKEPOCH：包含自己最后处理的事务id  和  self.currentEpoch    （为了和Leader进行数据同步）
            QuorumPacket ackNewEpoch = new QuorumPacket(Leader.ACKEPOCH, lastLoggedZxid, epochBytes, null);
            writePacket(ackNewEpoch, true);

            return ZxidUtils.makeZxid(newEpoch, 0);
        } else {
            if (newEpoch > self.getAcceptedEpoch()) {
                self.setAcceptedEpoch(newEpoch);
            }
            // 如果数据包类型不是 NEWLEADER ，则报错：Leader发来的第一个数据包，应该是 NEWLEADER 类型的。
            if (qp.getType() != Leader.NEWLEADER) {
                LOG.error("First packet should have been NEWLEADER");
                throw new IOException("First packet should have been NEWLEADER");
            }
            return qp.getZxid();
        }
    }

    /**
     * Finally, synchronize our history with the Leader.
     * @param newLeaderZxid
     * @throws IOException
     * @throws InterruptedException
     */
    protected void syncWithLeader(long newLeaderZxid) throws IOException, InterruptedException {
        QuorumPacket ack = new QuorumPacket(Leader.ACK, 0, null, null);
        long newEpoch = ZxidUtils.getEpochFromZxid(newLeaderZxid);

        // 已提交的事务
        LinkedList<Long> packetsCommitted = new LinkedList<Long>();
        // 尚未提交的事务
        LinkedList<PacketInFlight> packetsNotCommitted = new LinkedList<PacketInFlight>();

        // 读取数据包
        QuorumPacket qp = new QuorumPacket();
        readPacket(qp);

        // 在DIFF数据同步模式下，我们不需要处理快照，因为所有的事务将会被同步，在任意存在的快照文件顶部。
        // 对于SNAP和TRUNC情况：快照需要保存该历史记录。
        boolean snapshotNeeded = true;

        synchronized (zk) {
            if (qp.getType() == Leader.DIFF) { // 【增量数据同步】
                LOG.info("Getting a diff from the leader 0x{}", Long.toHexString(qp.getZxid()));
                // 设置标识位：标识当前数据同步不是快照同步模式。（在下面，创建完）
                snapshotNeeded = false;

            } else if (qp.getType() == Leader.SNAP) {// 【全量数据同步】
                LOG.info("Getting a snapshot from leader 0x" + Long.toHexString(qp.getZxid()));
                // The leader is going to dump the database
                // clear our own database and read
                zk.getZKDatabase().clear();
                zk.getZKDatabase().deserializeSnapshot(leaderIs);
                String signature = leaderIs.readString("signature");
                if (!signature.equals("BenWasHere")) {
                    LOG.error("Missing signature. Got " + signature);
                    throw new IOException("Missing signature");
                }
                zk.getZKDatabase().setlastProcessedZxid(qp.getZxid());

            } else if (qp.getType() == Leader.TRUNC) { // 【事务回滚】
                // 需要回滚事务日志到leader.lastzxid
                LOG.warn("Truncating log to get in sync with the leader 0x" + Long.toHexString(qp.getZxid()));
                boolean truncated = zk.getZKDatabase().truncateLog(qp.getZxid());
                if (!truncated) {
                    // not able to truncate the log
                    LOG.error("Not able to truncate the log " + Long.toHexString(qp.getZxid()));
                    System.exit(13);
                }
                zk.getZKDatabase().setlastProcessedZxid(qp.getZxid());
            } else {
                LOG.error("Got unexpected packet from leader " + qp.getType() + " exiting ... ");
                System.exit(13);

            }
            zk.createSessionTracker();

            long lastQueued = 0;

            // in Zab V1.0 (ZK 3.4+) we might take a snapshot when we get the NEWLEADER message, but in pre V1.0
            // we take the snapshot on the UPDATE message, since Zab V1.0 also gets the UPDATE (after the NEWLEADER)
            // we need to make sure that we don't take the snapshot twice.
            boolean isPreZAB1_0 = true;

            // ‘增量’数据同步，需要记录事务日志。‘全量’数据同步，不需记录事务日志。
            boolean writeToTxnLog = !snapshotNeeded;
            // we are now going to start getting transactions to apply followed by an UPTODATE
            outerLoop:
            while (self.isRunning()) {
                readPacket(qp);
                switch (qp.getType()) {

                    case Leader.PROPOSAL:
                        PacketInFlight pif = new PacketInFlight();
                        pif.hdr = new TxnHeader();
                        pif.rec = SerializeUtils.deserializeTxn(qp.getData(), pif.hdr);
                        if (pif.hdr.getZxid() != lastQueued + 1) {
                            LOG.warn("Got zxid 0x" + Long.toHexString(pif.hdr.getZxid()) + " expected 0x" + Long.toHexString(lastQueued + 1));
                        }
                        lastQueued = pif.hdr.getZxid();
                        packetsNotCommitted.add(pif);
                        break;

                    case Leader.COMMIT:
                        if (!writeToTxnLog) { // 如果不需要写到事务日志中：则直接zk.processTxn 处理事务，并且把proposal从packetsNotCommitted移除。
                            pif = packetsNotCommitted.peekFirst();
                            if (pif.hdr.getZxid() != qp.getZxid()) {
                                LOG.warn("Committing " + qp.getZxid() + ", but next proposal is " + pif.hdr.getZxid());
                            } else {
                                zk.processTxn(pif.hdr, pif.rec); // 处理事务
                                packetsNotCommitted.remove(); // 把proposal从packetsNotCommitted移除。
                            }
                        } else {
                            packetsCommitted.add(qp.getZxid());
                        }
                        break;

                    case Leader.INFORM:
                        // 只有Observer服务器才会收到此类型的packet消息。对这个消息的处理相当于 PROPOSAL 和 COMMMIT.
                        PacketInFlight packet = new PacketInFlight();
                        packet.hdr = new TxnHeader();
                        packet.rec = SerializeUtils.deserializeTxn(qp.getData(), packet.hdr);
                        // Log warning message if txn comes out-of-order
                        if (packet.hdr.getZxid() != lastQueued + 1) {
                            LOG.warn("Got zxid 0x" + Long.toHexString(packet.hdr.getZxid()) + " expected 0x" + Long.toHexString(lastQueued + 1));
                        }
                        lastQueued = packet.hdr.getZxid();
                        if (!writeToTxnLog) {
                            // Apply to db directly if we haven't taken the snapshot
                            zk.processTxn(packet.hdr, packet.rec);
                        } else {
                            packetsNotCommitted.add(packet);
                            packetsCommitted.add(qp.getZxid());
                        }
                        break;
                    case Leader.UPTODATE:
                        if (isPreZAB1_0) {
                            zk.takeSnapshot();
                            self.setCurrentEpoch(newEpoch);
                        }
                        self.cnxnFactory.setZooKeeperServer(zk);
                        break outerLoop;
                    case Leader.NEWLEADER: // Getting NEWLEADER here instead of in discovery
                        // means this is Zab 1.0
                        // Create updatingEpoch file and remove it after current
                        // epoch is set. QuorumPeer.loadDataBase() uses this file to
                        // detect the case where the server was terminated after
                        // taking a snapshot but before setting the current epoch.
                        File updating = new File(self.getTxnFactory().getSnapDir(), QuorumPeer.UPDATING_EPOCH_FILENAME);
                        if (!updating.exists() && !updating.createNewFile()) {
                            throw new IOException("Failed to create " + updating.toString());
                        }
                        if (snapshotNeeded) {
                            zk.takeSnapshot();
                        }
                        self.setCurrentEpoch(newEpoch);
                        if (!updating.delete()) {
                            throw new IOException("Failed to delete " + updating.toString());
                        }
                        writeToTxnLog = true; //Anything after this needs to go to the transaction log, not applied directly in memory
                        isPreZAB1_0 = false;
                        writePacket(new QuorumPacket(Leader.ACK, newLeaderZxid, null, null), true);
                        break;
                }
            }
        } // end  synchronized (zk)

        ack.setZxid(ZxidUtils.makeZxid(newEpoch, 0));
        writePacket(ack, true);
        sock.setSoTimeout(self.tickTime * self.syncLimit);
        zk.startup();
        /*
         * Update the election vote here to ensure that all members of the
         * ensemble report the same vote to new servers that start up and
         * send leader election notifications to the ensemble.
         *
         * @see https://issues.apache.org/jira/browse/ZOOKEEPER-1732
         */
        self.updateElectionVote(newEpoch);

        // We need to log the stuff that came in between the snapshot and the uptodate
        if (zk instanceof FollowerZooKeeperServer) {
            FollowerZooKeeperServer fzk = (FollowerZooKeeperServer) zk;
            for (PacketInFlight p : packetsNotCommitted) {
                fzk.logRequest(p.hdr, p.rec);
            }
            for (Long zxid : packetsCommitted) {
                fzk.commit(zxid);
            }
        } else if (zk instanceof ObserverZooKeeperServer) {
            // Similar to follower, we need to log requests between the snapshot
            // and UPTODATE
            ObserverZooKeeperServer ozk = (ObserverZooKeeperServer) zk;
            for (PacketInFlight p : packetsNotCommitted) {
                Long zxid = packetsCommitted.peekFirst();
                if (p.hdr.getZxid() != zxid) {
                    // log warning message if there is no matching commit
                    // old leader send outstanding proposal to observer
                    LOG.warn("Committing " + Long.toHexString(zxid) + ", but next proposal is " + Long.toHexString(p.hdr.getZxid()));
                    continue;
                }
                packetsCommitted.remove();
                Request request = new Request(null, p.hdr.getClientId(),
                                              p.hdr.getCxid(), p.hdr.getType(), null, null);
                request.txn = p.rec;
                request.hdr = p.hdr;
                ozk.commitRequest(request);
            }
        } else {
            // New server type need to handle in-flight packets
            throw new UnsupportedOperationException("Unknown server type");
        }
    }

    protected void revalidate(QuorumPacket qp) throws IOException {
        ByteArrayInputStream bis = new ByteArrayInputStream(qp.getData());
        DataInputStream dis = new DataInputStream(bis);
        long sessionId = dis.readLong();
        boolean valid = dis.readBoolean();
        ServerCnxn cnxn = pendingRevalidations.remove(sessionId);
        if (cnxn == null) {
            LOG.warn("Missing session 0x" + Long.toHexString(sessionId) + " for validation");
        } else {
            zk.finishSessionInit(cnxn, valid);
        }
        if (LOG.isTraceEnabled()) {
            ZooTrace.logTraceMessage(LOG,
                                     ZooTrace.SESSION_TRACE_MASK,
                                     "Session 0x" + Long.toHexString(sessionId) + " is valid: " + valid);
        }
    }

    protected void ping(QuorumPacket qp) throws IOException {
        // Send back the ping with our session data
        ByteArrayOutputStream bos = new ByteArrayOutputStream();
        DataOutputStream dos = new DataOutputStream(bos);
        HashMap<Long, Integer> touchTable = zk.getTouchSnapshot();
        for (Entry<Long, Integer> entry : touchTable.entrySet()) {
            dos.writeLong(entry.getKey());
            dos.writeInt(entry.getValue());
        }
        qp.setData(bos.toByteArray());
        writePacket(qp, true);
    }

    /**
     * Shutdown the Peer
     */
    public void shutdown() {
        // set the zookeeper server to null
        self.cnxnFactory.setZooKeeperServer(null);
        // clear all the connections
        self.cnxnFactory.closeAll();
        // shutdown previous zookeeper
        if (zk != null) {
            zk.shutdown();
        }
    }

    boolean isRunning() {
        return self.isRunning() && zk.isRunning();
    }

    static class PacketInFlight {
        TxnHeader hdr;
        Record rec;
    }
}
